package com.air.antispider.stream.dataprocess.businessprocess

import com.air.antispider.stream.common.util.jedis.{JedisConnectionUtil, PropertiesUtil}
import org.apache.spark.rdd.RDD
import org.json4s.DefaultFormats
import org.json4s.jackson.Json
import redis.clients.jedis.JedisCluster

/**
  * 链路统计功能
  */
object BusinessProcess {
  def linkCount(messageRDD: RDD[String]): RDD[(String, Int)] = {
    //数据采集量
    //String => (ip,1)
    val serverCountMapRDD: RDD[(String, Int)] = messageRDD.map(line => {
      val arr: Array[String] = line.split("#CS#")
      if (arr.length > 9) {
        //如果长度满足,那么就取IP
        val ip = arr(9)
        (ip, 1)
      } else {
        //切割后的长度不满足条件
        ("", 1)
      }
    })
    //开始聚合计算
    //reduceByKey((老的数据,新的数据) => 老的数据+新的数据)
    val serverCountReduce: RDD[(String, Int)] = serverCountMapRDD.reduceByKey(_+_)

    //当前活跃连接数
    val activeCountMapRDD: RDD[(String, Int)] = messageRDD.map(line => {
      val arr: Array[String] = line.split("#CS#")
      if (arr.length > 11) {
        //如果长度满足,那么就取数据
        val ip = arr(9)
        val num = arr(11).toInt
        (ip, num)
      } else {
        //切割后的长度不满足条件
        ("", 1)
      }
    })
    //只需要返回最新的当前活跃连接数就可以了
    val activeCountReduce: RDD[(String, Int)] = activeCountMapRDD.reduceByKey((x,y) => y)

    //将计算结果封装为Json,存入Redis
//    private Map<String,Integer> serversCountMap;
//    private Map<String,Integer> activeNumMap;

//{
//  serversCountMap:{ip,3},
//  activeNumMap:{ip,43}
//}
    if (!serverCountReduce.isEmpty() && !activeCountReduce.isEmpty()){
      //如果数据不为空,那么就取出数据
      val serversCountMap: collection.Map[String, Int] = serverCountReduce.collectAsMap()
      val activeNumMap: collection.Map[String, Int] = activeCountReduce.collectAsMap()

      //创建一个Map,用来封装最终的结果集
      var dataMap: Map[String, collection.Map[String, Int]] = Map[String, collection.Map[String, Int]]()
      dataMap += ("serversCountMap" -> serversCountMap)
      dataMap += ("activeNumMap" -> activeNumMap)

      //将map转换为Json
      val json: String = Json(DefaultFormats).write(dataMap)
//      println(json)
      //将json数据放入Redis
      val jedis: JedisCluster = JedisConnectionUtil.getJedisCluster
      val key: String = PropertiesUtil.getStringByKey("cluster.key.monitor.linkProcess", "jedisConfig.properties") + System.currentTimeMillis()
      val ex: Int = PropertiesUtil.getStringByKey("cluster.exptime.monitor", "jedisConfig.properties").toInt
      //设置数据
//      jedis.set(key, json)
      //设置数据的有效期为2分钟
      jedis.setex(key, ex, json)
    }

    serverCountMapRDD
  }

}
